package com.tj.bigdata.syncblock.strategy;

import com.tj.bigdata.syncblock.config.Web3jConfig;
import com.tj.bigdata.syncblock.entity.BlockInfo;
import com.tj.bigdata.syncblock.entity.ReceiptLog;
import com.tj.bigdata.syncblock.entity.Transactions;
import com.tj.bigdata.syncblock.exception.EsException;
import com.tj.bigdata.syncblock.service.EsUpService;
import io.reactivex.disposables.Disposable;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;
import org.web3j.protocol.Web3j;
import org.web3j.protocol.core.DefaultBlockParameter;
import org.web3j.protocol.core.methods.response.EthBlock;
import org.web3j.protocol.core.methods.response.TransactionReceipt;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.math.BigInteger;
import java.util.ArrayList;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * web3J同步方式
 *
 * @author guoch
 */
@Slf4j
@Service
public class Web3jSyncBlockImpl implements SyncBlockStrategy {

    @Autowired
    private EsUpService esUpService;

    @Autowired
    private ThreadPoolExecutor executorService;

    @Autowired
    private Web3jConfig web3jConfig;

    private Disposable subscribe;


    @Override
    public void batchBlockchain() {
        Web3j web3 = web3jConfig.getWeb3j();
        log.debug("连接以太坊成功");
        BigInteger maxBlockNumber = esUpService.maxBlockBumber() != null ? esUpService.maxBlockBumber().subtract(new BigInteger(String.valueOf(200))) : BigInteger.ZERO;
        subscribe =  web3.replayPastAndFutureBlocksFlowable((DefaultBlockParameter.valueOf(maxBlockNumber)), true)
                .doOnError(e -> log.error("on error:{}", e.getMessage()))
                .subscribe(this::executeBlock);
    }

    @Override
    public void batchBlockchain(BigInteger begin, BigInteger end) {

    }

    public void execute(Runnable r) {
        executorService.execute(r);
    }

    public void executeBlock(EthBlock block) {

        execute(() -> {
            // 获取到所需的块信息
            EthBlock.Block ethBlock = block.getBlock();
            BlockInfo blockInfo = new BlockInfo();
            BeanUtils.copyProperties(ethBlock, blockInfo);
            log.info("==========区块信息封装成功 {}====》", ethBlock.getNumber());
            if (Objects.nonNull(blockInfo)) {
                try {
                    Boolean blockUoRes = esUpService.blockInfoUp(blockInfo);
                    if (!blockUoRes){
                        throw new EsException(blockInfo.getNumber() + "区块信息上传ES失败,retry..");
                    }
                    log.info("==========区块信息上传ES成功 {}=======",blockInfo.getNumber());
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
            // transaction信息获取
            executeTransaction(ethBlock.getTransactions());
        });
    }

    public void executeTransaction(List<EthBlock.TransactionResult> transactions) {
        if (transactions.size() == 0) {
            return;
        }
        try {
            Web3j web3j = web3jConfig.getWeb3j();
            ArrayList<Transactions> transactionsList = new ArrayList<>();
            for (EthBlock.TransactionResult<EthBlock.TransactionObject> transactionResult : transactions) {
                EthBlock.TransactionObject transaction = transactionResult.get();
                Transactions transactionss = new Transactions();
                BeanUtils.copyProperties(transaction, transactionss);
                log.info("==========开始获取交易信息 {}==========", transactionss.getHash());
                // TransactionReceipt 数据获取
                TransactionReceipt receipt = web3j.ethGetTransactionReceipt(transaction.getHash()).send().getResult();
                log.info("==========交易回执获取成功 {}====》", receipt);
                BeanUtils.copyProperties(receipt, transactionss);
                ArrayList<ReceiptLog> receiptLogs = new ArrayList<>();
                receipt.getLogs().stream().forEach(log1 -> {
                    ReceiptLog receiptLog = new ReceiptLog();
                    receiptLog.setRemoved(log1.isRemoved()).setLogIndex(log1.getLogIndex())
                            .setAddress(log1.getAddress()).setData(log1.getData()).setTopics(log1.getTopics());
                    receiptLogs.add(receiptLog);
                });
                transactionss.setLogs(receiptLogs);
                transactionsList.add(transactionss);
            }
            if (!CollectionUtils.isEmpty(transactionsList)) {
                try {
                    Boolean transactionUpRes = esUpService.transactionInfoUp(transactionsList);
                    if (!transactionUpRes) {
                        throw new EsException(transactionsList.get(0).getBlockNumber() + "区块交易信息上传ES失败,retry..");
                    }
                    log.info("==========交易信息上传ES成功 {}=======",transactionsList.get(0).getBlockNumber());
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        } catch (IOException e) {
            log.error("transaction input error, msg:{}", e.getMessage());
        }
    }

    @PreDestroy
    public void preDestroy() {
        log.info("大数据 - 区块同步 - 结束");
    }
}
